package com.cloud.job.consumer;

import com.cloud.api.domain.job.VehicleLogin;
import com.cloud.job.doris.DorisStreamLoad;
import com.cloud.job.utils.DateUtils;
import com.cloud.job.utils.JsonUtils;
import com.fasterxml.jackson.databind.node.ObjectNode;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

import static com.cloud.job.utils.JsonUtils.*;

/**
 * @author 杨斌
 * 2024-12-2024-12-18
 */
@Slf4j
@Component
@ConfigurationProperties
public class MessageLoginConsumer {
    @Resource
    private DorisStreamLoad dorisStreamLoad;

    @KafkaListener(id="${listener.consumer2.id}",topics = "#{'${listener.consumer2.topics}'.split(',')}", groupId = "${listener.consumer2.groupId}",concurrency = "${listener.consumer2.concurrency}")
    public void processLoginMessage(@Payload List<ConsumerRecord<String, String>> consumerRecords,
                               Acknowledgment acknowledgment) {
        log.info("Starting to process login messages...");
        List<VehicleLogin> vehicleLoginList = new ArrayList<>();
        try {

            log.info("Polled {} login messages ",consumerRecords.size());
            if (consumerRecords.isEmpty()) {
                log.info("No login messages to process");
                return;
            }
            for (ConsumerRecord<String, String> record : consumerRecords) {
                try {
                    VehicleLogin vehicleLogin = parseMessage(record.value());
                    vehicleLoginList.add(vehicleLogin);
                } catch (Exception e) {
                    log.error("Error parsing login messages: {}", record.value(), e);
                }
            }
            if (vehicleLoginList.isEmpty()) {
                log.info("No parsed login messages to insert into database");
                return;
            }
            String jsonString = JsonUtils.writeValueAsString(vehicleLoginList);
            long loadStartTime = System.currentTimeMillis();
            dorisStreamLoad.loadJson(jsonString, VehicleLogin.class);
            long loadEndTime = System.currentTimeMillis();
            log.info("Inserted {} login messages records into database in {} ms", vehicleLoginList.size(), (loadEndTime - loadStartTime));
            // 提交偏移量
            acknowledgment.acknowledge();
            log.info("Committed offsets for processed login messages.");
        } catch (Exception e) {
            log.error("Error processing login messages", e);
        } finally {
            log.info("Finished processing login messages.");
        }
    }

    private VehicleLogin parseMessage(String message) throws Exception {
        ObjectNode objectNode = JsonUtils.readObjectNode(message);
        VehicleLogin.VehicleLoginBuilder vehicleLoginBuilder = VehicleLogin.builder();

        String loginTime = getTextValue(objectNode, "dataContent", "localDateTime");

        String dt = DateUtils.parseAndFormatToDt(loginTime);

        vehicleLoginBuilder
                .loginTime(loginTime)
                .dt(dt)
                .loginSerialNumber(getIntValue(objectNode, "dataContent", "vehicleLogin", "loginSerialNumber"))
                .vin(getTextValue(objectNode, "header", "vin"))
                .receiveTime(getTextValue(objectNode, "receive_time"))
                .iccid(getTextValue(objectNode, "dataContent", "vehicleLogin", "iccid"))
                .rechargeableSubsystemsCount(getShortValue(objectNode, "dataContent", "vehicleLogin", "rechargeableSubsystemsCount"))
                .rechargeableSubsystemsLength(getShortValue(objectNode, "dataContent", "vehicleLogin", "rechargeableSubsystemsLength"))
                .rechargeableSubsystemsCodes(getStringListValue(objectNode, "dataContent", "vehicleLogin", "rechargeableSubsystemsCodes"))
                .record(getTextValue(objectNode, "record"))
                .vehicleType(getTextValue(objectNode, "vehicle_type"));
        return vehicleLoginBuilder.build();
    }
}
